package com.xiaoyushu.count.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.xiaoyushu.count.biz.constant.MQConstants;
import com.xiaoyushu.count.biz.domain.mapper.UserCountDOMapper;
import com.xiaoyushu.framework.common.util.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 整合Guava 令牌桶削峰
 * 为了防止写库压力过大,在CountFans2DBConsumer 消费者类中整合一下 Guava令牌桶,每秒创建 5000个令牌,达到流量削峰的目的
 */
@Component
@RocketMQMessageListener(consumerGroup = "xiaoyushu_group_"+ MQConstants.TOPIC_COUNT_FANS_2_DB,
topic = MQConstants.TOPIC_COUNT_FANS_2_DB //主题 Topic
)
@Slf4j
public class CountFans2DBConsumer implements RocketMQListener<String> {
    @Resource
    private UserCountDOMapper userCountDOMapper;
    //每秒创建 5000个令牌
   private RateLimiter rateLimiter= RateLimiter.create(5000);



    /**
     * @param body
     */
    @Override
    public void onMessage(String body) {
        //流量削峰:通过获取令牌,如果没有令牌可用,将阻塞,知道获取
        rateLimiter.acquire();
        log.info("## 消费到了MQ [计数:粉丝数入库],{}...",body);

        Map<Long,Integer> countMap=null;
        try{
            countMap= JsonUtils.parseMap(body, Long.class,Integer.class);
        }catch ( Exception e){
            log.error("## 解析 JSON 字符串异常",e);
        }
        if (CollUtil.isNotEmpty(countMap)){
            //判断数据库中,若目标用户的记录不存在,则插入;若记录已经存在,则直接更新
            countMap.forEach((k,v)-> userCountDOMapper.insertOrUpdateFansTotalByUserId(v,k));
        }


    }
}
